import java.text.SimpleDateFormat
import java.util.Locale
import java.util.concurrent.TimeUnit

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Taxi {
  def main(args: Array[String]): Unit = {
    //创建sparkSession
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("Taxi")
      .getOrCreate()
    //导入隐式转换和函数
    import spark.implicits._
    import org.apache.spark.sql.functions._
    //数值读取
    val taxiRaw: Dataset[Row] = spark.read
      .option("header", true)
      .csv("E:\\data\\spark数据\\train.csv")
    //    taxiRaw.show(100)
    //    taxiRaw.printSchema()
    //转换操作
    val taxiParsed: RDD[Either[Trip, (Row, Exception)]] = taxiRaw.rdd.map(safe(parse))

    /**
      * 过滤出所有出异常的row
      * val result = taxiParsed.filter(e => e.isRight)
      * .map(e => e.right.get)
      */
    val taxiGood: Dataset[Trip] = taxiParsed.map(either => either.left.get).toDS()
    //绘制时长直方图  时长 = 开始时间的时间戳减去时间
    //编写UDF完成时长计算，将毫秒转换为小时
    val hours = (pickUpTime: Long, dropOffTime: Long) => {
      val duration = dropOffTime - pickUpTime
      val hours = TimeUnit.HOURS.convert(duration, TimeUnit.MILLISECONDS)
      hours
    }
    val hoursUDF = udf(hours)
    //5.2进行统计
    taxiGood.groupBy(hoursUDF($"pickUpTime", $"dropOffTime") as "duration")
      .count()
      .sort("duration")

    /**
      * 根据直方图，查看数据发布，剪除反常数据
      */
    spark.udf.register("hours",hours)
    val taxiClean: Dataset[Trip] = taxiGood.where("hours(pickUpTime,dropOffTime) between 0 and 3")
  }

  /**
    * 作用是封装parse方法，捕获异常
    *
    */
  def safe[P, R](f: P => R): P => Either[R, (P, Exception)] = {
    new Function[P, Either[R, (P, Exception)]] with Serializable {
      override def apply(param: P): Either[R, (P, Exception)] = {
        try {
          Left(f(param))
        } catch {
          case exception: Exception => Right((param, exception))
        }
      }
    }
  }

  /**
    * 将row转换成trip
    *
    * @param row
    */
  def parse(row: Row): Trip = {
    val richRow = new RichRow(row)
    val id = richRow.getAs[String]("id").orNull
    val pickUpTime = parseTime(richRow, "pickup_datetime")
    val dropOffTime = parseTime(richRow, "dropoff_datetime")
    val pickUpX = parseLocation(richRow, "pickup_longitude")
    val pickUpY = parseLocation(richRow, "pickup_latitude")
    val dropOffX = parseLocation(richRow, "dropoff_longitude")
    val dropOffY = parseLocation(richRow, "dropoff _longitude")
    Trip(id, pickUpTime, dropOffTime, pickUpX, pickUpY, dropOffX, dropOffY)
  }

  /**
    * def parseTime(row: RichRow, field: String): Long
    * 参数列表是row     类型是RichRow
    * filed   类型是String
    * 返回值的类型是  Long
    *
    * @param row
    * @param field
    * @return
    */
  def parseTime(row: RichRow, field: String): Long = {
    //1、表示时间类型的格式  SimpleDateFormat
    val pattern = "yyyy-MM-dd HH:mm:ss"
    val formatter = new SimpleDateFormat(pattern, Locale.ENGLISH)
    //2、执行转换，获取Date对象
    val time: Option[String] = row.getAs[String](field)
    //只有option中有值才会被调用
    val timeOption: Option[Long] = time.map(time => formatter.parse(time).getTime)
    //如果没有值返回0L
    timeOption.getOrElse(0L)

  }

  def parseLocation(row: RichRow, field: String): Double = {
    //获取数据
    val location = row.getAs[String](field)
    //转换数据
    val localtionOption = location.map(loc => loc.toDouble)
    //如果有值就是其本身，如果没有值那就是设定的值
    localtionOption.getOrElse(0.00)
  }
}

/**
  * DataFrame中的Row的包装类型，主要为了包装getAs方法
  *
  * @param row
  */
class RichRow(row: Row) {
  /**
    * 为了返回Option提醒外面处理空值，提供了处理方法
    *
    * @param field
    * @tparam T
    * @return
    */
  def getAs[T](field: String): Option[T] = {
    //判断是Row中对应的filed是否为空
    //null -- >None
    //not null -> st
    if (row.isNullAt(row.fieldIndex(field))) {
      None
    } else {
      Some(row.getAs[T](field))
    }
  }
}

case class Trip(
                 id: String,
                 pickUpTime: Long,
                 dropOffTime: Long,
                 pickUpX: Double,
                 pickUpY: Double,
                 dropOffX: Double,
                 dropOffY: Double
               )